package site.teamo.biu.flink.function.process;

import com.alibaba.fastjson2.JSON;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import site.teamo.biu.flink.sdk.annotation.function.BFProcessFunction;
import site.teamo.biu.flink.sdk.function.AbstractProcessFunction;
import site.teamo.biu.flink.sdk.job.JobContext;

/**
 * @author haocongshun
 * @date 2023/07/27 15:37:48
 */
@BFProcessFunction(
        out = String.class
)
public class ToJSONStringFunction<IN> extends AbstractProcessFunction<IN, String> {
    public ToJSONStringFunction(String name, JobContext context) {
        super(name, context);
    }

    @Override
    public void processElement(IN in, ProcessFunction<IN, String>.Context context, Collector<String> collector) throws Exception {
        collector.collect(JSON.toJSONString(in));
    }
}
